package com.bjbn;

import com.bjbn.bean.entity.FltInfoDynamic;
import com.bjbn.service.FlightHandle;
import com.bjbn.service.FltInfoDynamicService;
import com.bjbn.service.FlightDataProcessor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class FlightDataConsumers {
    /*
    融合总线_imf_航班航站动态信息_输出	csb_imf_d_fltairline_out
    融合总线_imf_航班值机柜台信息_输出	csb_imf_d_fltcounter_out
    融合总线_imf_航班登机门信息_输出	csb_imf_d_fltgate_out
    融合总线_imf_航班动态信息_输出	csb_imf_d_fltinfo_out
    融合总线_imf_航班动态保障节点信息_输出	csb_imf_d_flt_sg_dyn_out
    融合总线_imf_航班行李提取转盘信息_输出	csb_imf_d_fltlugeftch_out
    融合总线_imf_航班行李滑槽口信息_输出	csb_imf_d_fltlugload_out
    融合总线_imf_共享航班信息_输出	csb_imf_d_fltshare_out
    融合总线_imf_航班机位信息_输出	csb_imf_d_fltstand_out
    融合总线_航班旅客和行李信息_输出	csb_d_flt_passenger_out
    */

    @Value("${flight.data.kafka.target-topic}")
    private String targetTopic;

    @Autowired
    private FltInfoDynamicService fltInfoDynamicService;

    @Autowired
    private FlightDataProcessor flightDataProcessor;
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @KafkaListener(topics = "csb_imf_d_fltairline_out")
    public void consumeAirlineInfo(String message) {
        // 使用专门的处理器来处理航班航站信息，避免阻塞Kafka消费者
        flightDataProcessor.processAirlineInfo(message);
    }

    @KafkaListener(topics = "csb_imf_d_fltcounter_out")
    public void consumeCounterInfo(String message) {
        // Process counter info
        // System.out.println("Received csb_imf_d_fltcounter_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltCounter(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }
    }

    @KafkaListener(topics = "csb_imf_d_fltgate_out")
    public void consumeGateInfo(String message) {
        // Process gate info
        // System.out.println("Received csb_imf_d_fltgate_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltGate(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }

    }


    @KafkaListener(topics = "csb_imf_d_fltinfo_out")
    public void consumeFlightInfo(String message) {
        // Process flight info
        // System.out.println("Received csb_imf_d_fltinfo_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltInfo(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }
    }

    @KafkaListener(topics = "csb_imf_d_flt_sg_dyn_out")
    public void consumeFlightSgDynInfo(String message) {
        // Process flight sg dyn info
        // System.out.println("Received csb_imf_d_flt_sg_dyn_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltSgDyn(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }
    }

    @KafkaListener(topics = "csb_imf_d_fltlugeftch_out")
    public void consumeFlightLugeftchInfo(String message) {
        // Process flight lugeftch info
        // System.out.println("Received csb_imf_d_fltlugeftch_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltLugeftch(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }
    }

    @KafkaListener(topics = "csb_imf_d_fltlugload_out")
    public void consumeFlightLugloadInfo(String message) {
        // Process flight lugload info
        // System.out.println("Received csb_imf_d_fltlugload_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltLugload(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }
    }

    @KafkaListener(topics = "csb_imf_d_fltshare_out")
    public void consumeFlightShareInfo(String message) {
        // Process flight share info
        // System.out.println("Received csb_imf_d_fltshare_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltShare(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }
    }

    @KafkaListener(topics = "csb_imf_d_fltstand_out")
    public void consumeFlightStandInfo(String message) {
        // Process flight stand info
        // System.out.println("Received csb_imf_d_fltstand_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handleFltStand(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }
    }

    @KafkaListener(topics = "csb_d_flt_passenger_out")
    public void consumePassengerInfo(String message) {
        // Process flight passenger info
        // System.out.println("Received csb_d_flt_passenger_out message: " + message);

        FltInfoDynamic flightInfo = FlightHandle.handlePassenger(message);
        if (null != flightInfo) {
            log.debug("FlightInfo: {}", flightInfo);
            fltInfoDynamicService.saveOrUpdateFltInfoDynamic(flightInfo);
            // 将航班动态数据推送到Kafka
            sendToKafka(flightInfo);
        }

    }
    
    /**
     * 将航班动态数据发送到Kafka
     * @param flightInfo 航班动态信息
     */
    private void sendToKafka(FltInfoDynamic flightInfo) {
        try {
            // 使用航班ID作为消息键，确保相同航班的数据分区一致
            String key = flightInfo.getFLID().toString();
            kafkaTemplate.send(targetTopic, key, flightInfo);
            log.debug("Successfully sent flight data to Kafka topic {}. Flight ID: {}", targetTopic, key);
        } catch (Exception e) {
            log.error("Failed to send flight data to Kafka. Error: {}", e.getMessage(), e);
        }
    }

}